package com.aotain.nyx.statis;

import java.util.HashMap;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;

import com.aotain.nyx.ipstat.IPStatTuple;


/**
 * 
     *@Title:异常攻击类型统计 Map
     *@Description:
     *@Author:Turk
     *@Since:2017年7月5日
     *@Version:1.1.0
 */
public class AbnStatisMap implements MapFunction<AbnStatisTuple, Tuple2<String,IPStatTuple>> {

	/**
	 * 
	 */
	private static final long serialVersionUID = -2251740822545426679L;

	@Override
	public Tuple2<String, IPStatTuple> map(AbnStatisTuple value)
			throws Exception {
		// TODO Auto-generated method stub
		Tuple2<String,IPStatTuple> ret = new Tuple2<String,IPStatTuple>();
		if(value != null) {
			String dip = value.getDestIP();	
			//
			
			IPStatTuple stat = new IPStatTuple();
			HashMap<String,Long> valueMap = new HashMap<String,Long>();			
			valueMap.put(value.getType(), value.getUpStreamOctets());
			valueMap.put(value.getType() + "NUM", value.getUpStreamPacket());
			valueMap.put(value.getType() + "ATTACKNUM", value.getAttackNum());
			valueMap.put("ATTACKNUM", value.getAttackNum());//攻击次数，一次完整的攻击
			stat.setDestIP(value.getDestIP());
			stat.setValueMap(valueMap);
			
			//异常总体统计
			valueMap.put("ATTOCT", value.getUpStreamOctets());
			valueMap.put("ATTPACKET", value.getUpStreamPacket());
			stat.setAttackMax(value.getAttackMax());
			ret.setFields(dip, stat);
		}
		return ret;
	}


}
